package com.zwh.rocketmq.demo.transaction;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;

import java.util.Map;

/**
 * 注册事务监听器，
 * 相当于：rocketMQTemplate.getProducer().setTransactionListener(bean)
 * @author zwh
 * @date 2024-1-22 14:22
 */
@RocketMQTransactionListener
@Component
public class CustomTransactionListener implements RocketMQLocalTransactionListener {
    @Autowired
    private TransactionalUtil transactionalUtil;
    /**
     * 半事务消息发送成功后，执行本地事务
     * @param msg
     * @param arg 这里接收到了参数是发送成功后 sendMessageInTransaction(destination, message1, arg)的arg参数
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        TransactionStatus status = null;
        try {
            // 事务开始
            status = transactionalUtil.begin();
            // 数据库操作
            Map<String, Object> orderMap = (Map<String, Object>)arg;
            // 获取header头部存的数据
            String orderId = (String) msg.getHeaders().get(RocketMQHeaders.KEYS);
            int result = 1; // orderMapper.addOrder(orderEntity);
            // 本地事务提交
            transactionalUtil.commit(status);
            if (result < 1) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            if (status != null) {
                transactionalUtil.rollback(status);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        return null;
    }

    /**
     * 若服务端未收到发送者提交的二次确认结果，或服务端收到的二次确认结果为Unknown未知状态，经过固定时间后，
     * MQ服务端对生产者实例发起消息回查。
     * 服务端仅仅会按照参数尝试指定次数，超过次数后事务会强制回滚
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        /*
        生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
        生产者根据检查得到的本地事务的最终状态再次提交二次确认
         */
        String transactionId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        String orderId = (String) msg.getHeaders().get(RocketMQHeaders.KEYS);
        if (transactionId == null || orderId == null) {
            // rollback则，半消息不会投递到队列，而是删除
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        //查询数据库
        Object orderEntity1 = null;// orderMapper.findOrder(orderId);
        if (null == orderEntity1) {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }
}
